package cn.enjoy.rocketmq.component;

import cn.enjoy.rocketmq.prop.AliyunMqProperties;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentLinkedQueue;

@Slf4j
@Component
public class MqComponent {
    @Autowired
    private ProducerBean producerBean;
    @Autowired
    private AliyunMqProperties aliyunMqProperties;

    private static final ConcurrentLinkedQueue<Message> ALL_EXCEPTION_QUEUE = new ConcurrentLinkedQueue<>();
    private static final ConcurrentLinkedQueue<Message> ALL_ASYNC_EXCEPTION_QUEUE = new ConcurrentLinkedQueue<>();
    private static final ConcurrentLinkedQueue<Message> MP_EXCEPTION_QUEUE = new ConcurrentLinkedQueue<>();


    /**
     * 发送异步消息-使用默认ProducerBean
     *
     * @param bytes jackson转的字节数组
     * @param tag   标签
     */
    public void sendAsyncMessage(byte[] bytes, String tag) {
        sendAsyncMessage(bytes, tag, 0);
    }

    /**
     * 发送异步消息-使用默认ProducerBean
     *
     * @param bytes       jackson转的字节数组
     * @param tag         标签
     * @param delaySecond 延迟时间 分钟
     */
    public void sendAsyncMessage(byte[] bytes, String tag, int delaySecond) {
        Message msg = new Message(aliyunMqProperties.getTopic(), tag, bytes);
        if (delaySecond > 0) {
            long delayTime = System.currentTimeMillis() + ((long) delaySecond * 60 * 1000);
            msg.setStartDeliverTime(delayTime);
        }
        sendAsyncResult(MP_EXCEPTION_QUEUE, producerBean, "sendMqMessage", msg);
    }

    /**
     * 发送异步消息
     *
     * @param errMqQueue 错误消息列表
     * @param producer   生产者
     * @param queueName  队列名称
     * @param msg        消息
     */
    private void sendAsyncResult(
            ConcurrentLinkedQueue<Message> errMqQueue,
            ProducerBean producer,
            String queueName,
            Message msg) {

        if (producer.isClosed()) {
            producer.start();
        }
        //处理异常的消息 重发消息
        processAsyncException(errMqQueue, producer, queueName);

        try {
            //发送消息
            producer.sendAsync(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info(">>> {}.消息发送成功！tag:{}", queueName, msg.getTag());
                }

                @Override
                public void onException(OnExceptionContext context) {
                    log.error(">>> {}.消息发送失败！tag:{},ex:{}", queueName,
                            msg.getTag(), context.getException());
                    errMqQueue.add(msg);
                }
            });

        } catch (Exception e) {
            log.error(">>> queueName:{},ex:{}", queueName, e);
            errMqQueue.add(msg);
        }
    }

    /**
     * 重发一次列表中的错误消息，可以选择不使用
     *
     * @param queue     错误消息列表
     * @param producer  生产者
     * @param queueName 队列名称
     */
    private void processAsyncException(ConcurrentLinkedQueue<Message> queue, ProducerBean producer, String queueName) {
        if (queue.size() == 0) {
            return;
        }
        Message lastExceptionMsg = null;
        try {
            lastExceptionMsg = queue.poll();
            if (lastExceptionMsg != null) {
                Message finalLastExceptionMsg = lastExceptionMsg;
                producer.sendAsync(lastExceptionMsg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.error(">>> {}.消息重发成功！tag:{}", queueName, finalLastExceptionMsg.getTag());
                    }

                    @Override
                    public void onException(OnExceptionContext context) {
                        log.error(">>> {}.消息重发失败！tag:{},ex:{}", queueName,
                                finalLastExceptionMsg.getTag(), context.getException());
                        queue.add(finalLastExceptionMsg);
                    }
                });
            }
        } catch (Exception e) {
            log.error(">>> {}.allQueue:{}", queueName, e);
            queue.add(lastExceptionMsg);
        }
    }

    /**
     * 发送异步消息-指定producer
     *
     * @param producer 生产者
     * @param topic    主题
     * @param tag      标签
     * @param bytes    jackson转的字节数组
     */
    public void sendAsyncMessage(ProducerBean producer, String topic, String tag, byte[] bytes) {
        sendAsyncMessage(producer, topic, tag, bytes, 0);
    }

    /**
     * 发送异步消息-指定producer
     *
     * @param producer 生产者
     * @param topic    主题
     * @param tag      标签
     * @param bytes    jackson转的字节数组
     * @param min      延迟时间 分钟
     */
    public void sendAsyncMessage(ProducerBean producer, String topic, String tag, byte[] bytes, int min) {

        Message msg = new Message(topic, tag, bytes);

        if (min > 0) {
            long delayTime = System.currentTimeMillis() + (min * 60 * 1000);
            msg.setStartDeliverTime(delayTime);
        }
        sendAsyncResult(ALL_ASYNC_EXCEPTION_QUEUE, producer, "sendAllAsyncMqMessage", msg);
    }

    /**
     * 发送同步消息
     *
     * @param producer 生产者
     * @param topic    主题
     * @param tag      标签
     * @param bytes    jackson转的字节数组
     * @return SendResult
     */
    public SendResult sendMessage(ProducerBean producer, String topic, String tag, byte[] bytes) {
        return sendMessage(producer, topic, tag, bytes, 0);
    }

    /**
     * 发送同步消息
     *
     * @param producer 生产者
     * @param topic    主题
     * @param tag      标签
     * @param bytes    jackson转的字节数组
     * @param min      延迟时间 分钟
     * @return SendResult
     */
    public SendResult sendMessage(ProducerBean producer, String topic, String tag, byte[] bytes, int min) {

        Message msg = new Message(topic, tag, bytes);

        if (min > 0) {
            long delayTime = System.currentTimeMillis() + (min * 60 * 1000);
            msg.setStartDeliverTime(delayTime);
        }
        return sendResult(ALL_EXCEPTION_QUEUE, producer, "sendAllMqMessage", msg);
    }

    /**
     * 发送同步消息
     */
    private SendResult sendResult(
            ConcurrentLinkedQueue<Message> errMqQueue,
            ProducerBean producer,
            String queueName,
            Message msg) {


        if (producer.isClosed()) {
            producer.start();
        }
        //处理异常的消息 重发消息
        processLastException(errMqQueue, producer, queueName);

        SendResult result = null;
        try {
            //发送消息
            result = producer.send(msg);
        } catch (Exception e) {
            log.error(">>> queueName:{},ex:{}", queueName, e);
            errMqQueue.add(msg);
        }

        return result;
    }

    /**
     * 当消息发送失败,重新发送
     */
    private void processLastException(ConcurrentLinkedQueue<Message> queue, ProducerBean producer, String queueName) {
        if (queue.size() == 0) {
            return;
        }
        Message lastExceptionMsg = null;
        try {
            lastExceptionMsg = queue.poll();
            if (lastExceptionMsg != null) {
                producer.send(lastExceptionMsg);
                log.error(">>> {}.消息重发成功！tag:{}", queueName, lastExceptionMsg.getTag());
            }
        } catch (Exception e) {
            log.error(">>> {}.allQueue:{}", queueName, e);
            queue.add(lastExceptionMsg);
        }
    }

    /**
     * 发送单向消息 不管结果
     *
     * @param producer 生产者
     * @param topic    主题
     * @param tag      标签
     * @param bytes    jackson转的字节数组
     */
    public void sendOneWayMessage(ProducerBean producer, String topic, String tag, byte[] bytes) {
        sendOneWayMessage(producer, topic, tag, bytes, 0);
    }

    /**
     * 发送单向消息 不管结果
     *
     * @param producer 生产者
     * @param topic    主题
     * @param tag      标签
     * @param bytes    jackson转的字节数组
     * @param min      延迟时间 分钟
     */
    public void sendOneWayMessage(ProducerBean producer, String topic, String tag, byte[] bytes, int min) {

        Message msg = new Message(topic, tag, bytes);

        if (min > 0) {
            long delayTime = System.currentTimeMillis() + (min * 60 * 1000);
            msg.setStartDeliverTime(delayTime);
        }
        sendOneWayResult(producer, "sendAllOneWayMqMessage", msg);
    }

    /**
     * 发送单向消息
     */
    private void sendOneWayResult(ProducerBean producer, String queueName, Message msg) {

        if (producer.isClosed()) {
            producer.start();
        }

        try {
            //发送消息
            producer.sendOneway(msg);
        } catch (Exception e) {
            log.error(">>> queueName:{},ex:{}", queueName, e);
        }
    }

}
